package com.hxh.rpc.consumer;

import com.hxh.rpc.model.RpcRequest;
import com.hxh.rpc.model.RpcResponse;
import com.hxh.rpc.registry.ServiceRegistry;
import com.hxh.rpc.registry.model.ServiceMetaConfig;
import com.hxh.rpc.util.ServiceUtils;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;

/**
 * @author hxh
 * @date 2022/3/2
 * @since 1.0
 */
@Slf4j
public class RpcConsumerNettyHandler extends SimpleChannelInboundHandler<RpcResponse> {

    @Autowired
    private ServiceRegistry serviceRegistry;
    /**
     * 消费者锁
     */
    private final Object lock = new Object();

    private RpcResponse rpcResponse;
    /**
     * netty EventLoopGroup
     */
    private EventLoopGroup eventLoopGroup = new NioEventLoopGroup(4);
    private Channel channel;

    /**
     * 发起rpc请求
     *
     * @param rpcRequest
     */
    public RpcResponse sendRpcRequest(RpcRequest rpcRequest) {
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline()
                                .addLast(
                                        new ObjectDecoder(ClassResolvers.weakCachingResolver(this.getClass().getClassLoader())),
                                        new ObjectEncoder(),
                                        new RpcConsumerNettyHandler())
                        ;

                    }
                });
        String key = ServiceUtils.buildServiceKey(rpcRequest.getClassName(), rpcRequest.getServiceVersion());
        try {
            ServiceMetaConfig serviceMetaConfig = this.serviceRegistry.discovery(key);
            if (serviceMetaConfig == null) {
                throw new Exception("send rpcRequest error, serviceMetaConfig is null");
            }
            log.info("sendRpcRequest begin: {}", serviceMetaConfig);
            ChannelFuture channelFuture = bootstrap.connect(serviceMetaConfig.getHost(), serviceMetaConfig.getPort()).sync();
            channelFuture.addListener((future) -> {
                if (channelFuture.isSuccess()) {
                    log.info("rpc invoke success");
                } else {
                    log.info("rpc invoke error", channelFuture.cause());
                    eventLoopGroup.shutdownGracefully();
                }
            });
            this.channel = channelFuture.channel();
            this.channel.writeAndFlush(rpcRequest).sync();
            synchronized (this.lock) {
                log.info("sendRpcRequest lock.wait");
                this.lock.wait();
            }
            log.info("get rpc response=" + rpcResponse.toString());
            return this.rpcResponse;
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            return null;
        } finally {
            //关闭相关连接
            if (this.channel != null) {
                this.channel.close();
            }
            if (this.eventLoopGroup != null) {
                this.eventLoopGroup.shutdownGracefully();
            }
        }

    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcResponse rpcResponse) {
        this.rpcResponse = rpcResponse;
        //收到远程服务接口的响应数据，通知调用端
        synchronized (lock) {
            log.info("lock.notifyAll");
            lock.notifyAll();
        }
    }

}
